package org.apache.rocketmq.namesrv;

import org.apache.commons.lang3.concurrent.BasicThreadFactory;
import org.apache.rocketmq.common.ThreadFactoryImpl;
import org.apache.rocketmq.common.constant.LoggerName;
import org.apache.rocketmq.common.future.FutureTaskExt;
import org.apache.rocketmq.common.namesrv.NamesrvConfig;
import org.apache.rocketmq.common.utils.ThreadUtils;
import org.apache.rocketmq.logging.org.slf4j.Logger;
import org.apache.rocketmq.logging.org.slf4j.LoggerFactory;
import org.apache.rocketmq.namesrv.processor.ClientRequestProcessor;
import org.apache.rocketmq.namesrv.processor.DefaultRequestProcessor;
import org.apache.rocketmq.namesrv.route.ZoneRouteRPCHook;
import org.apache.rocketmq.namesrv.routeinfo.BrokerHousekeepingService;
import org.apache.rocketmq.namesrv.routeinfo.RouteInfoManager;
import org.apache.rocketmq.remoting.Configuration;
import org.apache.rocketmq.remoting.RemotingClient;
import org.apache.rocketmq.remoting.RemotingServer;
import org.apache.rocketmq.remoting.netty.*;
import org.apache.rocketmq.remoting.protocol.RequestCode;

import java.util.concurrent.*;



/**
 * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
 * @author：陈清风扬，个人微信号：chenqingfengyangjj。
 * @date:2024/12/25
 * @方法描述：启动Namesrv模块的核心类，该类会初始化Namesrv模块的各个组件，然后启动Namesrv的服务端
 */
public class NamesrvController {

    private static final Logger LOGGER = LoggerFactory.getLogger(LoggerName.NAMESRV_LOGGER_NAME);

    private static final Logger WATER_MARK_LOG = LoggerFactory.getLogger(LoggerName.NAMESRV_WATER_MARK_LOGGER_NAME);

    //Namesrv要用到的配置信息成员变量
    private final NamesrvConfig namesrvConfig;

    //Netty服务端要用到的配置信息成员变量
    private final NettyServerConfig nettyServerConfig;

    //Netty客户端要用到的配置信息成员变量
    private final NettyClientConfig nettyClientConfig;

    //Namesrv的服务器
    private RemotingServer remotingServer;

    //Namesrv的客户端
    private RemotingClient remotingClient;

    //该成员变量会持有Namesrv模块用到的所有配置信息
    private final Configuration configuration;

    //这个成员变量会监听Broker和Namesrv构建的channel状态变化
    private final BrokerHousekeepingService brokerHousekeepingService;

    //定时任务执行器，这个定时任务执行器是专门用来执行打印服务端接收请求和处理请求详细信息的任务的
    private final ScheduledExecutorService scheduledExecutorService = ThreadUtils.newScheduledThreadPool(1,
            new BasicThreadFactory.Builder().namingPattern("NSScheduledThread").daemon(true).build());

    //这个定时任务执行器是专门用来扫描Namesrv内部保存的不活跃的Broker的channel的
    private final ScheduledExecutorService scanExecutorService = ThreadUtils.newScheduledThreadPool(1,
            new BasicThreadFactory.Builder().namingPattern("NSScanScheduledThread").daemon(true).build());

    //这个执行器是专门处理DefaultRequestProcessor处理器的请求的
    private ExecutorService defaultExecutor;

    //这个客户端执行器，是专门处理生产者和消费者客户端发送过来的请求的
    private ExecutorService clientRequestExecutor;

    //defaultExecutor执行器要使用的任务队列
    private BlockingQueue<Runnable> defaultThreadPoolQueue;

    //clientRequestExecutor执行器要使用的任务队列
    private BlockingQueue<Runnable> clientRequestThreadPoolQueue;

    //路由信息管理器
    private final RouteInfoManager routeInfoManager;


    public NamesrvController(NamesrvConfig namesrvConfig, NettyServerConfig nettyServerConfig) {
        this(namesrvConfig, nettyServerConfig, new NettyClientConfig());
    }

    //构造方法
    public NamesrvController(NamesrvConfig namesrvConfig, NettyServerConfig nettyServerConfig, NettyClientConfig nettyClientConfig) {
        this.namesrvConfig = namesrvConfig;
        this.nettyServerConfig = nettyServerConfig;
        this.nettyClientConfig = nettyClientConfig;
        this.brokerHousekeepingService = new BrokerHousekeepingService(this);
        this.routeInfoManager = new RouteInfoManager(namesrvConfig, this);
        //创建了Configuration成员变量，并且把namesrvConfig、nettyServerConfig配置信息交给configuration成员变量管理了
        this.configuration = new Configuration(LOGGER, this.namesrvConfig, this.nettyServerConfig);
        //把namesrv.properties配置文件的默认路径赋值给configuration的storePathField成员变量
        this.configuration.setStorePathFromConfig(this.namesrvConfig, "configStorePath");
    }

    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @date:2024/12/25
     * @方法描述：初始化Namesrv要用到的各个组件的方法
     */
    public boolean initialize() {

        //初始化网络组件，在第一版本代码中其实就是Netty构建的Namesrv服务器
        initiateNetworkComponents();
        //初始化执行器组件
        initiateThreadExecutors();
        //注册请求处理器到Netty构建的服务端中
        registerProcessor();
        //启动定时任务执行器
        startScheduleService();
        //初始化RPC操作的回调对象
        initiateRpcHooks();
        return true;
    }


    //初始化网络组件的方法
    private void initiateNetworkComponents() {
        //创建Netty构建的服务器对象
        this.remotingServer = new NettyRemotingServer(this.nettyServerConfig, this.brokerHousekeepingService);
        //创建Namesrv的客户端
        this.remotingClient = new NettyRemotingClient(this.nettyClientConfig);
    }


    //启动定时任务执行器的方法，在源码中，这个方法中实际上还启动了scanExecutorService执行器
    //让这个执行器定期检测不活跃的Broker的channel，如果有不活跃的channel就关闭该channel
    //但在我为大家提供的第二版本代码中，这里只是简单启动scheduledExecutorService执行器工作，这个定时任务执行器会定时打印服务端处理请求的水位线信息
    private void startScheduleService() {

        //定期扫描不活跃的Broker的channel，这个操作到第六版本代码才使用，第五版版暂时注释掉

        this.scanExecutorService.scheduleAtFixedRate(NamesrvController.this.routeInfoManager::scanNotActiveBroker,
                5, this.namesrvConfig.getScanNotActiveBrokerInterval(), TimeUnit.MILLISECONDS);
        this.scheduledExecutorService.scheduleAtFixedRate(() -> {
            try {//每隔一秒打印一次水位线信息
                NamesrvController.this.printWaterMark();
            } catch (Throwable e) {
                LOGGER.error("printWaterMark error.", e);
            }
        }, 10, 1, TimeUnit.SECONDS);
    }


    //打印服务端处理请求的水位线信息的方法
    private void printWaterMark() {
        //其实就是把服务端处理请求的任务队列中的待处理的请求的数量，以及请求的慢速时间打印出来，具体逻辑请看headSlowTimeMills方法
        WATER_MARK_LOG.info(
                "[WATERMARK] ClientQueueSize:{} ClientQueueSlowTime:{} " +
                        "DefaultQueueSize:{} DefaultQueueSlowTime:{}",
                //这里获得的是存放客户端请求的任务队列中待处理请求的数量
                this.clientRequestThreadPoolQueue.size(),
                headSlowTimeMills(this.clientRequestThreadPoolQueue),
                //这里得到的是存放默认处理器接收到的待处理的请求的数量
                this.defaultThreadPoolQueue.size(),
                headSlowTimeMills(this.defaultThreadPoolQueue)
        );
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @date:2024/12/26
     * @方法描述：计算待处理的请求的慢速时间的方法，实际上是这样的，当Namesrv服务端接收到客户端请求后，会根据请求的类型，把请求交给ClientRequestProcessor客户端请求处理器
     * 或者是DefaultRequestProcessor默认请求处理器来处理，而处理器在处理请求的时候并不会直接就处理，而是把请求存放到处理器使用的执行器的任务队列中，执行器会不断处理任务队列中的请求
     * 这个时候我们就可以取出任务队列中的第一个任务，等到这个任务被存放到队列多长时间了，因为队列中的第一个任务肯定是最早被存放进队列但是一直没有被处理的任务
     * 通过该任务等待被处理的时间，我们就可以大概判断出当前服务端处理请求的效率，这个时间就是慢速时间，慢速时间越长，说明服务端处理请求的效率越低，这个时候就需要采取一些措施提高服务端处理请求的效率
     * 最简单直接的就是增加线程池现成的数量
     */
    private long headSlowTimeMills(BlockingQueue<Runnable> q) {
        long slowTimeMills = 0;
        //得到任务队列中第一个任务
        final Runnable firstRunnable = q.peek();
        if (firstRunnable instanceof FutureTaskExt) {
            final Runnable inner = ((FutureTaskExt<?>) firstRunnable).getRunnable();
            //要被处理的请求都会被包装成RequestTask任务
            if (inner instanceof RequestTask) {
                //当前时间减去任务被创建的时间，得到的就是任务在队列中存放的时间，也就是慢速时间
                slowTimeMills = System.currentTimeMillis() - ((RequestTask) inner).getCreateTimestamp();
            }
        }
        if (slowTimeMills < 0) {
            slowTimeMills = 0;
        }
        //返回慢速时间，这个时间要交给printWaterMark方法打印
        return slowTimeMills;
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @date:2024/12/26
     * @方法描述：初始化执行器组件的方法
     */
    private void initiateThreadExecutors() {
        //创建defaultExecutor执行器要使用的任务队列
        this.defaultThreadPoolQueue = new LinkedBlockingQueue<>(this.namesrvConfig.getDefaultThreadPoolQueueCapacity());
        //创建defaultExecutor执行器，这个执行器会被DefaultRequestProcessor请求处理器使用
        this.defaultExecutor = ThreadUtils.newThreadPoolExecutor(
                this.namesrvConfig.getDefaultThreadPoolNums(),
                this.namesrvConfig.getDefaultThreadPoolNums(),
                1000 * 60,
                TimeUnit.MILLISECONDS,
                this.defaultThreadPoolQueue,
                new ThreadFactoryImpl("RemotingExecutorThread_")
        );
        //创建clientRequestExecutor执行器要使用的任务队列
        this.clientRequestThreadPoolQueue = new LinkedBlockingQueue<>(this.namesrvConfig.getClientRequestThreadPoolQueueCapacity());
        //创建clientRequestExecutor执行器，这个执行器会被ClientRequestProcessor请求处理器使用
        this.clientRequestExecutor = ThreadUtils.newThreadPoolExecutor(
                this.namesrvConfig.getClientRequestThreadPoolNums(),
                this.namesrvConfig.getClientRequestThreadPoolNums(),
                1000 * 60,
                TimeUnit.MILLISECONDS,
                this.clientRequestThreadPoolQueue,
                new ThreadFactoryImpl("ClientRequestExecutorThread_")
        );
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @date:2024/12/26
     * @方法描述：注册请求处理器到Netty构建的服务端的方法
     */
    private void registerProcessor() {
        //创建客户端请求处理器，这个客户端请求处理器专门处理客户端根据主题获取路由信息的请求
        ClientRequestProcessor clientRequestProcessor = new ClientRequestProcessor(this);
        //把该处理器注册到服务端中，这里可以看到该处理器使用clientRequestExecutor执行器处理请求
        //RequestCode.GET_ROUTEINFO_BY_TOPIC就是该处理器处理的请求类型
        this.remotingServer.registerProcessor(RequestCode.GET_ROUTEINFO_BY_TOPIC, clientRequestProcessor, this.clientRequestExecutor
        );
        //注册默认请求处理器到服务端中，默认请求处理器使用的defaultExecutor执行器
        this.remotingServer.registerDefaultProcessor(
                new DefaultRequestProcessor(this),
                this.defaultExecutor
        );

    }


    //初始化RPC操作回调对象的方法
    private void initiateRpcHooks() {
        //把RPC回调对象注册到Namesrv的服务器中
        this.remotingServer.registerRPCHook(new ZoneRouteRPCHook());
    }

    //启动NamesrvController控制器的方法
    public void start() throws Exception {
        //启动Namesrv服务器
        this.remotingServer.start();
        if (0 == nettyServerConfig.getListenPort()) {
            nettyServerConfig.setListenPort(this.remotingServer.localListenPort());
        }

        this.routeInfoManager.start();
    }

    public void shutdown() {
        this.remotingServer.shutdown();
        this.defaultExecutor.shutdown();
        this.clientRequestExecutor.shutdown();
        this.scheduledExecutorService.shutdown();
        this.scanExecutorService.shutdown();
        this.routeInfoManager.shutdown();
    }

    public NamesrvConfig getNamesrvConfig() {
        return namesrvConfig;
    }

    public NettyServerConfig getNettyServerConfig() {
        return nettyServerConfig;
    }

    public void setRemotingServer(RemotingServer remotingServer) {
        this.remotingServer = remotingServer;
    }

    public Configuration getConfiguration() {
        return configuration;
    }

    public RouteInfoManager getRouteInfoManager() {
        return routeInfoManager;
    }

    public RemotingClient getRemotingClient() {
        return remotingClient;
    }
}
